package com.shujia.ads

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.catalog.hive.HiveCatalog


/*
  统计评论中的男女比例
  需要 dwd_kafka_comment_msk 与 dwd_kafka_user_msk 关联
  然后按gender分组，做count
 */
object Demo1ComputeGenderIndex {
  def main(args: Array[String]): Unit = {
    val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()

    val bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)

    val hiveCatalog = new HiveCatalog("myHive", "sent", "SentimentCompute/src/main/resources")

    bsTableEnv.registerCatalog("myHive", hiveCatalog)
    bsTableEnv.useCatalog("myHive")

    bsTableEnv.executeSql(
      """
        |insert into ads.ads_mysql_gender_index
        |select
        |t2.gender
        |,count(*) as c
        |from dwd.dwd_kafka_comment_msk t1
        |inner join dwd.dwd_kafka_user_msk t2
        |on t1.user_id = t2.id
        |group by t2.gender
        |""".stripMargin)

    bsEnv.execute()

  }

}
